package online.chazi.bootplus.commons.executor;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.*;

/**
 * 说明：作者很懒，什么都没留下
 * Created by 叉子同学 on 2020-04-30 15:22
 */
@Slf4j
public class PooledBlockExecutor<T> {

    private int maxPoolSize = 200;
    private BlockingQueue<Runnable> queue;
    private ThreadPoolExecutor executor;

    private long historyMax = 0;
    private long debugTime = 0;

    public PooledBlockExecutor() {
        queue = new ArrayBlockingQueue<>(maxPoolSize);
        executor = new ThreadPoolExecutor(1, maxPoolSize, 60, TimeUnit.SECONDS, queue, new ThreadPoolExecutor.AbortPolicy());
    }

    public PooledBlockExecutor(int maxPoolSize) {
        this.maxPoolSize = maxPoolSize;
        queue = new ArrayBlockingQueue<>(maxPoolSize);
        executor = new ThreadPoolExecutor(1, maxPoolSize, 60, TimeUnit.SECONDS, queue, new ThreadPoolExecutor.AbortPolicy());
    }

    private boolean isFull() {
        return queue.size() >= maxPoolSize;
    }

    private void trace() {
        int size = queue.size();
        if (size > historyMax) historyMax = size;
        long now = System.currentTimeMillis();
        if (now - debugTime > 500) {
            debugTime = now;
            log.debug("Executor Queue Size={}, historyMax={}", size, historyMax);
        }
    }

    private void sleep() {
        try {
            Thread.sleep(100L);
        } catch (InterruptedException e) {
            //
        }
    }

    public void execute(ExecuteWorker<T> worker) {

        while (isFull()) {
            sleep();
        }

        trace();

        boolean reject;
        do {
            reject = false;
            try {
                executor.execute(worker);
            } catch (RejectedExecutionException e) {
                log.warn("满啦满啦。{}", e.getMessage());
                reject = true;
                while (isFull()) {
                    sleep();
                }
            }
        } while (reject);

    }

}
